package net.xuele.learn.flux;

import org.junit.Test;
import reactor.core.publisher.Flux;
import reactor.core.scheduler.Schedulers;

import java.time.Duration;
import java.time.temporal.ChronoUnit;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Random;

/**
 *
 * Reactive Programming 作为观察者模式（Observer） 的延伸，不同于传统的命令编程方式（ Imperative programming）同步拉取数据的方式，
 * 如迭代器模式（Iterator） 。而是采用数据发布者同步或异步地推送到数据流（Data Streams）的方案。
 * 当该数据流（Data Steams）订阅者监听到传播变化时，立即作出响应动作。
 * 在实现层面上，Reactive Programming 可结合函数式编程简化面向对象语言语法的臃肿性，屏蔽并发实现的复杂细节，
 * 提供数据流的有序操作，从而达到提升代码的可读性，以及减少 Bugs 出现的目的。
 * 同时，Reactive Programming 结合背压（Backpressure）的技术解决发布端生成数据的速率高于订阅端消费的问题。
 *
 * https://www.ibm.com/developerworks/cn/java/j-cn-with-reactor-response-encode/index.html
 */
public class FluxDemo {

    public static void main(String[] args) throws InterruptedException {
        Flux.just("hello", "world").subscribe(System.out::println);
        Flux.fromArray(new Integer[]{1, 2, 3}).subscribe(System.out::println);
        Flux.empty().subscribe(System.out::println);
        System.out.println("=====================");
        Flux.range(8, 10).subscribe(System.out::println);
        System.out.println("-------从0开始生成数据，每隔1s生成一个数据-------");
        //
        Flux.interval(Duration.of(1, ChronoUnit.SECONDS)).subscribe(System.out::println);
        Thread.sleep(10000);
        // 毫秒的时间间隔级别
        System.out.println("---------");
        Flux.interval(Duration.ofMillis(1000)).subscribe(System.out::println);
    }

    @Test
    public void generate() {
        // synchronousSink的next方法只能被对应一次，多次调用会报错
        Flux.generate(synchronousSink -> {
            synchronousSink.next("hello");
            synchronousSink.complete();
        }).subscribe(System.out::println);
        System.out.println("-------上面说的next方法只能被调用一次指的是在某次generate中只能对应一次--------");
        final Random random = new Random();
        Flux.generate(ArrayList::new, (list, sink) -> {
            int value = random.nextInt(100);
            list.add(value);
            sink.next(value);
            if (list.size() == 10) {
                sink.complete();
            }
            return list;
        }).subscribe(System.out::println);
    }

    @Test
    public void create() {
        // create方法使用FluxSink，它支持同步和异步的消息产生，并且可以在一次调用中产生多个元素
        Flux.create(fluxSink -> {
            for (int i = 0; i < 10; i++) {
                fluxSink.next(i);
            }
            fluxSink.complete();
        }).subscribe(System.out::println);
    }

    @Test
    public void bufferAndFilter() {
        // 20为最大个数，可以生成5个集合，每个集合的长度为20
        Flux.range(0, 100).buffer(20).subscribe(System.out::println);
        // Flux序列每隔100毫秒生成一个，bufferMillis每隔1001毫秒收集一次，
        // 使用toStream转换成Stream是因为Flux序列是异步的，这么做可以保证主线程生成完成之后不会推出，从而可以输出所有的元素
//        Flux.interval(Duration.ofMillis(100)).bufferMillis(1001).take(2).toStream().forEach(System.out::println);
        // bufferUntil会一直收集直到 Predicate 返回为 true，收集了5个集合，每个集合的长度为2
        Flux.range(1, 10).bufferUntil(i -> i % 2 == 0).subscribe(System.out::println);
        // bufferWhile只会在满足条件时才会收集，收集了5个集合，每个集合的长度为1
        Flux.range(1, 10).bufferWhile(i -> i % 2 == 0).subscribe(System.out::println);
        // 过滤，满足条件才会保留
        Flux.range(1, 10).filter(s -> s % 2 == 0).subscribe(System.out::println);
    }

    /**
     * window 操作符的作用类似于 buffer，所不同的是 window 操作符是把当前流中的元素收集到另外的 Flux 序列中，因此返回值类型是 Flux<Flux<T>>
     */
    @Test
    public void window() {
        // 输出5个UnicastProcessor字符
        // 因为 window 操作符所产生的流中包含的是 UnicastProcessor 类的对象，而 UnicastProcessor 类的 toString 方法输出的就是 UnicastProcessor 字符
        Flux.range(1, 100).window(20).subscribe(System.out::println);
//        Flux.intervalMillis(100).windowMillis(1001).take(2).toStream().forEach(System.out::println);
    }

    /**
     * zipWith 操作符把当前流中的元素与另外一个流中的元素按照一对一的方式进行合并
     */
    @Test
    public void zipWith() {
        // 在合并时可以不做任何处理，由此得到的是一个元素类型为 Tuple2 的流，输出：[a,c][b,d]
        Flux.just("a", "b", "t").zipWith(Flux.just("c", "d", "u"))
                .subscribe(System.out::println);
        // 也可以通过一个 BiFunction 函数对合并的元素进行处理（默认是使用逗号），所得到的流的元素类型为该函数的返回值
        Flux.just("a", "b")
                .zipWith(Flux.just("c", "d"), (s1, s2) -> String.format("%s--%s", s1, s2))
                .subscribe(System.out::println);
    }

    /**
     * 用来从当前流中提取元素
     */
    @Test
    public void take() {
        Flux.range(1, 1000).takeUntil(i -> i == 5).subscribe(System.out::println);
    }

    /**
     * 对流中包含的所有元素进行累积操作
     * 在操作时可以指定一个初始值(reduceWith)。如果没有初始值，则序列的第一个元素作为初始值(reduce)
     */
    @Test
    public void reduce() {
        Flux.range(1, 100).reduce((x, y) -> x + y).subscribe(System.out::println);
    }

    /**
     * merge 和 mergeSequential 操作符用来把多个流合并成一个 Flux 序列
     * <p>
     * 不同之处在于merge按照所有流中元素的实际产生顺序来合并，而mergeSequential则按照所有流被订阅的顺序，以流为单位进行合并
     */
    @Test
    public void merge() {
        // 第一个Flux序列每隔100毫秒产生一个，而第二个则是延迟50毫秒。
        // 这两个序列的元素按照时间顺序交织在一起
//        Flux.merge(Flux.intervalMillis(0, 100).take(5), Flux.intervalMillis(50, 100).take(5))
//                .toStream()
//                .forEach(System.out::println);
//        // 先产生第一个流中的全部元素，再产生第二个流中的全部元素
//        Flux.mergeSequential(Flux.intervalMillis(0, 100).take(5), Flux.intervalMillis(50, 100).take(5))
//                .toStream()
//                .forEach(System.out::println);
    }

    /**
     * flatMap和flatMapSequential操作符把流中的每个元素转换成一个流，再把所有流中的元素进行合并。
     * flatMapSequential 和 flatMap 之间的区别与 mergeSequential 和 merge 之间的区别是一样的
     */
    @Test
    public void flatMap() {
        // 流中的元素被转换成每隔100毫秒产生的数量不同的流，再进行合并。
        // 由于第一个流中包含的元素数量较少，所以在结果流中一开始是两个流的元素交织在一起，然后就只有第二个流中的元素
//        Flux.just(5, 10)
//                .flatMap(x -> {
//                    System.out.println("----------" + x);
//                    return Flux.intervalMillis(x * 10, 100).take(x);
//                })
//                .toStream()
//                .forEach(System.out::println);
    }


    /**
     * concatMap 操作符的作用也是把流中的每个元素转换成一个流，再把所有流进行合并。
     * 与 flatMap 不同的是，concatMap 会根据原始流中的元素顺序依次把转换之后的流进行合并；
     * 与 flatMapSequential 不同的是，concatMap 对转换之后的流的订阅是动态进行的，而 flatMapSequential 在合并之前就已经订阅了所有的流
     */
    @Test
    public void concatMap() {
//        Flux.just(5, 10)
//                .concatMap(x -> {
//                    System.out.println("----------" + x);
//                    return Flux.intervalMillis(x * 10, 100).take(x);
//                })
//                .toStream()
//                .forEach(System.out::println);
    }

    /**
     * 把所有流中的最新产生的元素合并成一个新的元素，作为返回结果流中的元素
     * 只要其中任何一个流中产生了新的元素，合并操作就会被执行一次，结果流中就会产生新的元素
     */
    @Test
    public void combineLatest() {
//        Flux.combineLatest(Arrays::toString,
//                Flux.intervalMillis(100).take(5),
//                Flux.intervalMillis(50, 100).take(5)
//        ).toStream().forEach(System.out::println);
    }


    @Test
    public void Schedulers() {
        Flux.create(fluxSink -> {
            fluxSink.next(Thread.currentThread().getName());
            System.out.println("+++++++++++"+Thread.currentThread().getName());
            fluxSink.complete();
        })
                .publishOn(Schedulers.single())
                .map(x -> {
                    System.out.println("-----------"+x);
                    System.out.println("--------------"+Thread.currentThread().getName()+"---------");
                    return String.format("[%s] %s", Thread.currentThread().getName(), x);
                })
                .publishOn(Schedulers.elastic())
                .map(x -> {
                    System.out.println("==========="+x);
                    System.out.println("=============="+Thread.currentThread().getName()+"===============");
                    return String.format("[%s] %s", Thread.currentThread().getName(), x);
                })
                .subscribeOn(Schedulers.parallel())
                .toStream()
                .forEach(System.out::println);
    }
}
